package com.sui.bigdata.flink.sql.function.aggregate;

import com.sui.bigdata.flink.sql.function.scalar.NotBlank;
import com.sui.bigdata.flink.sql.function.util.PhoneHHIAccum;
import org.apache.flink.table.functions.AggregateFunction;

import java.math.BigDecimal;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.commons.lang3.ArrayUtils;

/**
 * @author LiuHuan
 * @date 2019/9/5 12:32
 * @description
 * @reviewer
 */
//通讯录异常号码占比
public class UserknPhoneUnusualRate extends AggregateFunction<Double, PhoneHHIAccum> {
    @Override
    public PhoneHHIAccum createAccumulator() {
        return new PhoneHHIAccum();
    }

    @Override
    public Double getValue(PhoneHHIAccum phoneHHIAccum) {
        Double result = 0.0000;
        Integer count = 0;
        if (!(phoneHHIAccum == null || phoneHHIAccum.getCount() == 0 || phoneHHIAccum.getGroups().isEmpty())) {
            Set<Map.Entry<String, Integer>> entries = phoneHHIAccum.getGroups().entrySet();
            for (Map.Entry<String, Integer> entry : entries
                    ) {
                count += entry.getValue();
            }
        }
        if (phoneHHIAccum.getCount() > 0) {
            Double newResult = (double) count / phoneHHIAccum.getCount();
            if (newResult != null && newResult >= 0) {
                BigDecimal bigDecimal = new BigDecimal(newResult);
                result = bigDecimal.setScale(4, BigDecimal.ROUND_HALF_UP).doubleValue();
            }
        }
        return result;
    }

    public void accumulate(PhoneHHIAccum phoneHHIAccum, String phone) {
        String[] array = {"2", "3", "5", "6", "7", "8"};
        if (NotBlank.eval(phone) && !phoneHHIAccum.exitsPhone(phone)) {
            Integer num;
            phoneHHIAccum.addphone(phone);
            phoneHHIAccum.setCount(phoneHHIAccum.getCount() + 1);
            //异常则为1 合法则为0
            if (isMobileNO(phone)) {
                if (phone.length() != 11 && "1".equals(phone.subSequence(0, 1))) {
                    num = 1;
                } else if ("0".equals(phone.subSequence(0, 1)) && (phone.length() > 12 || phone.length() < 11)) {
                    num = 1;
                } else if (phone.length() > 10 && "400".equals(phone.subSequence(0, 3))) {
                    num = 1;
                } else if (ArrayUtils.contains(array, phone.subSequence(0, 1).toString())) {
                    num = 1;
                } else if (phone.length() > 5 && "9".equals(phone.subSequence(0, 1))) {
                    num = 1;
                } else {
                    num = 0;
                }
            } else {
                num = 1;
            }
            phoneHHIAccum.getGroups().put(phone, num);
        }
    }

    public static boolean isMobileNO(String mobiles) {
        Pattern p = Pattern.compile("^[-\\+]?[\\d]*$");
        if (mobiles != null) {
            Matcher m = p.matcher(mobiles);
            return m.matches();
        }
        return false;
    }

    public void merge(PhoneHHIAccum phoneHHIAccum, Iterable<PhoneHHIAccum> it) {
        Iterator<PhoneHHIAccum> iter = it.iterator();
        while (iter.hasNext()) {
            PhoneHHIAccum a = iter.next();
            phoneHHIAccum.setGroups(a.getGroups());
            phoneHHIAccum.setCount(a.getCount());
            phoneHHIAccum.setRepetition(a.getRepetition());
        }
    }
}
